package com.teemor.gateway.filters.factory;

import com.alibaba.fastjson.JSON;
import com.teemor.common.bean.utils.IPInfoUtils;
import com.teemor.common.bean.utils.kafka.BaseKafkaProducer;
import com.teemor.gateway.openfeign.BlogServerFeign;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.gateway.filter.GatewayFilter;
import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.util.UriComponentsBuilder;

import java.net.URI;
import java.util.HashMap;
import java.util.Map;

/**
 * @Description:
 * @author zhoulk
 * @date: 2020/3/19
 */
@Component
@Slf4j
public class RecordIPGatewayFilterFactory extends AbstractGatewayFilterFactory<Object> {

    @Autowired
    private BlogServerFeign blogServerFeign;

    @Override
    public GatewayFilter apply(Object config) {
        return (exchange, chain) -> {
            ServerHttpRequest request = exchange.getRequest();
            String ip = IPInfoUtils.getIp(request);
            String path = exchange.getRequest().getURI().getPath();
            String method = request.getMethod().toString();
            String param = JSON.toJSONString(request.getQueryParams());
            request = request.mutate().uri(buildNewUri(exchange.getRequest().getURI(), ip)).build();
            sendToKafka(ip, path, method, param);
            return chain.filter(exchange.mutate().request(request).build());
        };
    }

    private URI buildNewUri(URI uri, String ip) {
        StringBuilder query = new StringBuilder();
        String originalQuery = uri.getRawQuery();
        if (StringUtils.hasText(originalQuery)) {
            query.append(originalQuery);
            if (originalQuery.charAt(originalQuery.length() - 1) != '&') {
                query.append('&');
            }
        }
        query.append("ip=");
        query.append(ip);
        try {
            return UriComponentsBuilder.fromUri(uri)
                    .replaceQuery(query.toString())
                    .build(true)
                    .toUri();
        } catch (RuntimeException ex) {
            return uri;
        }
    }


    @Async
    public void sendToKafka(String ip, String path, String method, String param) {
     try {
         Map<String, String> map = new HashMap<>();
         map.put("ip", ip);
         map.put("path", path);
         map.put("date", System.currentTimeMillis() + "");
         map.put("method", method);
         map.put("param", param);
         blogServerFeign.addIprecord(map);
//        ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>("topic-iprecord", JSON.toJSONString(map));
//        BaseKafkaProducer.sendMessage(record);
     }catch (Exception e) {
         log.error("record ip failed", e);
     }
    }
}
